package com.lzx.kafka.producer;

import com.lzx.kafka.handler.KafkaSendResultHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;


@Component
public class KfkaProducer {

	final KafkaTemplate<String, String> kafkaTemplate;

	@Autowired
	KafkaSendResultHandler producerListener;

	public KfkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
		this.kafkaTemplate = kafkaTemplate;
	}


	//为kafka添加事务
	@Transactional
	//发送消息方法
	public void send(String flag) throws ExecutionException, InterruptedException, TimeoutException {

		//消息结果回调配置
		kafkaTemplate.setProducerListener(producerListener);
		this.doSent("topic1", "11111111111");
		//默认同步发送


		if ("0".equals(flag)) {
			int a = 1 / 0;
		}
		this.doSent("topic2", "2222222");


        /*//同步发送   当send方法耗时大于get方法所设定的参数时会抛出一个超时异常，但需要注意，这里仅抛出异常，消息还是会发送成功的
        kafkaTemplate.send(topic, "test send message timeout").get(1, TimeUnit.MICROSECONDS);*/

		//异步发送 发送时间较长的时候会导致进程提前关闭导致无法调用回调函数
		Thread.sleep(1000);
	}

	private void doSent(String topic, String data) {
		System.out.println("我是生产者, 我往: " + topic + "里面发送了数据,数据是:   " + data);
		//默认同步发送
		ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, data);

	}


}
